package cn.doitedu.marketing.engine.querier;

import cn.doitedu.marketing.engine.beans.EventBean;
import cn.doitedu.marketing.engine.beans.EventCombinationCondition;
import cn.doitedu.marketing.engine.beans.EventCondition;
import cn.doitedu.marketing.engine.utils.EventUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.ListState;

import java.util.Iterator;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/7/25
 **/
@Slf4j
public class StateQuerier implements RuleConditionQuerier{
    ListState<EventBean> listState;
    public StateQuerier(ListState<EventBean> listState){
        this.listState = listState;
    }

    public int findEventCombinationConditionCount(String deviceId,EventCombinationCondition eventCombinationCondition, long st, long ed) throws Exception {
        String s = getEventCombinationStr(eventCombinationCondition, st, ed);
        Pattern r = Pattern.compile(eventCombinationCondition.getMatchPattern());
        Matcher matcher = r.matcher(s);
        int count = 0;
        while (matcher.find()) count++;

        log.debug("正则表达式为： {} ,state 返回的字符串为： {} ,匹配次数为：{}",eventCombinationCondition.getMatchPattern(), s, count);
        return count;
    }

    public String getEventCombinationStr(EventCombinationCondition eventCombinationCondition, long st, long ed) throws Exception {
        Iterator<EventBean> beans = listState.get().iterator();
        List<EventCondition> eventConditionList = eventCombinationCondition.getEventConditionList();
        StringBuilder sb = new StringBuilder();
        while (beans.hasNext()) {
            EventBean next = beans.next();
            if (next.getTimeStamp() >= st && next.getTimeStamp() <= ed) {
                for (int i = 1; i <= eventConditionList.size(); i++) {
                    if (EventUtil.eventMatchCondition(next,eventConditionList.get(i - 1))) {
                        sb.append(i + "");
                    }
                }
            }
        }

        return sb.toString();
    }

}
